package com.iflytek.uoamp.agent.monitor.sender.loadbalance;

import com.iflytek.uoamp.agent.monitor.mcagent.SendClient;
import com.iflytek.uoamp.agent.monitor.sender.config.Config;
import com.iflytek.uoamp.agent.monitor.sender.config.HostInfo;
import com.iflytek.uoamp.agent.monitor.sender.interfaces.ISender;
import com.iflytek.uoamp.agent.monitor.sender.interfaces.ISenderFactory;

import org.apache.commons.lang3.exception.ExceptionUtils;
import org.apache.log4j.Logger;

import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;

public class LoadbalanceManager implements Runnable {

	private final List<HostInfo> hostServers;
	// connectionPoolManager容器对象无需同步，从ConnectionPoolManager中取出的client需要同步
//	private final Map<String, ConnectionPoolManager> connectionPoolManager;
	private final Random random = new Random();
	private final ExecutorService callTimeoutPool;
	private Config config;
	private ISender client = SendClient.getInstance();
	/**
	 * 获取下一个host。只会获取到存活的host，并且经过负载均衡处理过的地址
	 *
	 * */
	private HostInfo nextHostServerInfo() {
		HostInfo retHostInfo = null;
		List<HostInfo> currentOKItems = new LinkedList<HostInfo>();
		List<Integer> currentOKItemsWeight = new LinkedList<Integer>();
		int totalWeight = 0;
		for (HostInfo host : hostServers) {
			if (host.isCanUse()) {
				currentOKItems.add(host);
				currentOKItemsWeight.add(host.getHostWeight());
			}
		}

		if (currentOKItems.size() != 0) {
			for (int i = 0; i < currentOKItemsWeight.size(); i++) {
				totalWeight += currentOKItemsWeight.get(i);
				currentOKItemsWeight.set(i, totalWeight);
			}
			int randomWeight = this.random.nextInt() % totalWeight + 1;
			for (int i = 0; i < currentOKItemsWeight.size(); i++) {
				if (currentOKItemsWeight.get(i) >= randomWeight) {
					retHostInfo = currentOKItems.get(i);
					break;
				}
			}
		}
		currentOKItems.clear();
		currentOKItemsWeight.clear();
		currentOKItems = null;
		currentOKItemsWeight = null;
		return retHostInfo;
	}

	/**
	 * 检查链接是否存活
	 * */
	private Future<Boolean> doPing(final ISender client) {
		if (null == this.callTimeoutPool) {
			return null;
		}
		return callTimeoutPool.submit(new Callable<Boolean>() {
			@Override
			public Boolean call() throws Exception {
				try {
					return client.ping();
				} catch (Exception e) {
					Logger.getRootLogger().info("Check ping call exception!!");
				}
				return false;
			}
		});
	}

	/**
	 * 检查主机是否存活
	 * */
	private void checkAvailable(HostInfo host) {
//		ConnectionPoolManager connPool = this.connectionPoolManager.get(host.Identity());
//		if (null == connPool) {
//			return;
//		}
//		ISender client = null;
		try {
//			client = connPool.checkout();
			if (null == client) {
				// 获取不到client对象，不代表client连接状态有问题，有可能是比较繁忙而没有获取到client连接对象，所以此处不应该关闭连接
				Logger.getRootLogger().debug("checkAvailable request sendClient failed.");
				return;
			}

			boolean isAlive = false;
			Future<Boolean> future = doPing(client);
			if (null != future) {
				try {
					isAlive = future.get(config.getPingTimeOut(), TimeUnit.MILLISECONDS);
				} catch (Exception e) {
					Logger.getRootLogger().info("Check ping result timeout!!");
				}
			}
			
			Logger.getRootLogger().info("DEBUG | check ping and status = " + isAlive);
			
			setHostServerEnabled(host, isAlive);
		} catch (Exception e) {
			Logger.getRootLogger().error(ExceptionUtils.getStackTrace(e), e);
		} finally {
//			if (null != client) {
//				connPool.checkIn(client);
//			}
		}
	}

	/**
	 * 检查主机是否存活
	 * */
	private void checkAvailable() {
//		ConnectionPoolManager connPool = this.connectionPoolManager.get(host.Identity());
//		if (null == connPool) {
//			return;
//		}
//		ISender client = null;
		try {
//			client = connPool.checkout();
			if (null == client) {
				// 获取不到client对象，不代表client连接状态有问题，有可能是比较繁忙而没有获取到client连接对象，所以此处不应该关闭连接
				Logger.getRootLogger().debug("checkAvailable request sendClient failed.");
				return;
			}

			boolean isAlive = false;
			Future<Boolean> future = doPing(client);
			if (null != future) {
				try {
					isAlive = future.get(config.getPingTimeOut(), TimeUnit.MILLISECONDS);
				} catch (Exception e) {
					Logger.getRootLogger().info("Check ping result timeout!!");
				}
			}
			
			Logger.getRootLogger().info("DEBUG | check ping and status = " + isAlive);
			
		} catch (Exception e) {
			Logger.getRootLogger().error(ExceptionUtils.getStackTrace(e), e);
		}
	}
	
	private void setHostServerEnabled(HostInfo unablehost, boolean enabled) {
		if (enabled != unablehost.isCanUse()) {
			unablehost.setCanUse(enabled);
			Logger.getRootLogger().info("agent status changed, now status : " + unablehost.Identity() + " : " + enabled);
		}
	}

	/**
	 * 获取具体的链接 会调用nextHostServerInfo 此函数有可能返回空
	 * */
//	public ConnectionPoolManager getNextConnectManager() {
//		ConnectionPoolManager connManager = null;
		// 从多个Server中获取一个可以正常通信的server信息
//		HostInfo curHostInfo = nextHostServerInfo();
//		if (curHostInfo != null) {
//			connManager = this.connectionPoolManager.get(curHostInfo.Identity());
//		}
//		return connManager;
//	}

	/**
	 * 目前没有调用
	 * */
	public void closeAll() throws Exception {
//		if (this.connectionPoolManager == null) {
//			throw new Exception("LoadbalanceManagerForSafeModel has been closed!");
//		}
//		
//		Set<Map.Entry<String, ConnectionPoolManager>> set = this.connectionPoolManager.entrySet();
//		Iterator<Map.Entry<String, ConnectionPoolManager>> it = set.iterator();
//		Map.Entry<String, ConnectionPoolManager> entry = null;
//		while (it.hasNext()) {
//			entry = it.next();
//			entry.getValue().closeAll();
//		}
//		this.connectionPoolManager.clear();
		this.callTimeoutPool.shutdown();
	}

	public LoadbalanceManager(ISenderFactory<ISender> factory, Config config) {
		this.config = config;
		this.hostServers = config.getHostInfo();
//		this.connectionPoolManager = new HashMap<String, ConnectionPoolManager>();
//		for (HostInfo host : hostServers) {
//			this.connectionPoolManager.put(host.Identity(), new ConnectionPoolManager(factory, config, host));
//		}
		final AtomicLong threadCounter = new AtomicLong(0);
		if (config.getPingIntervalTime() != -1) {
			this.lock = new byte[0];
			this.callTimeoutPool = Executors.newFixedThreadPool(1, new ThreadFactory() {
				@Override
				public Thread newThread(Runnable r) {
					Thread t = new Thread(r);
					t.setName("Netty ping thread pool - " + String.valueOf(threadCounter.incrementAndGet()));
					Logger.getRootLogger().info("Make a new ping thread : " + t.getName());
					return t;
				}
			});
		} else {
			this.lock = null;
			this.callTimeoutPool = null;
		}
	}

	private final byte[] lock;
	/**
	 * 休眠一段时间后检查链接是否存活。异常已经捕获，如果不检查很可能是死锁
	 * */
	@Override
	public void run() {
		if (null == this.callTimeoutPool) {
			return;
		}
		while (true) {
			try {
//				for (HostInfo host : hostServers) {
//					checkAvailable(host);
//				}
				checkAvailable();
				synchronized (lock) {
					lock.wait(config.getPingIntervalTime());
				}
			} catch (Exception e) {
				Logger.getRootLogger().error(ExceptionUtils.getStackTrace(e), e);
			}
		}
	}
}
